package com.br.aiops.receiver;

import com.alibaba.fastjson.JSONObject;
import com.br.aiops.counter.Counter;
import com.br.aiops.utils.DateUtils;
import lombok.Builder;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * kafka告警接收器
 */
@Component
public class AlertLogKafkaReceiver {

    private static final Logger LOGGER = LoggerFactory.getLogger(AlertLogKafkaReceiver.class);

    public static final String FIELD_MESSAGE = "message";
    public static final String FIELD_HOST = "host";
    public static final String FIELD_HOST_NAME = "name";
    public static final String FIELD_SOURCE = "source";
    public static final String FIELD_FIELDS = "fields";
    public static final String FIELD_SUB_ENV = "subEnv";

    public static final String VALUE_NONE_TIMESTAMP = "none-timestamp";
    public static final String VALUE_NONE_MSG = "none-msg";
    public static final String VALUE_NONE_LEVEL = "none-level";
    public static final String VALUE_NONE_ERROR_CODE = "none-error-code";
    public static final String VALUE_NONE_MSG_CONTENT = "none-msg-content";
    public static final String VALUE_NONE_SUBJECT_CONTENT = "none-subject-content";
    public static final String VALUE_NONE_RECEIVER_CONTENT = "none-receiver-content";
    public static final String VALUE_NONE_HOSTNAME = "none-hostname";
    public static final String RDS_CHAR_DOLLAR = "RDS_CHAR_DOLLAR";
    public static final String SPITE = "/";


    @Autowired
    private Counter counter;

    @Resource(name = "threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @KafkaListener(topics = "${spring.kafka.consumer.topic}")
    public void startConsume(ConsumerRecord<String, String> consumer) {
        LOGGER.info("aiops-log-alert收到消息：{}", consumer.value());
        String json = consumer.value();
        count(json, consumer.partition());
    }

    /**
     * 对输入的字符串进行统计：输入的字符串json中包括：level、errorCode字段
     * @param json
     */
    public void count(String json, int partition) {
        //使用线程池
        threadPoolTaskExecutor.execute(() -> {
            try {
                AlertLog alertLog = parse(json);

                //十分钟之前的消息直接丢弃不处理
                Calendar calendar = Calendar.getInstance();
                calendar.add(Calendar.MINUTE, -10);
                if (alertLog.getLogTime().before(calendar.getTime())) {
                    LOGGER.warn("消息为10分钟之前的消息，丢弃不处理。消息内容：{}", json);
                    return;
                }
                alertLog.setPartition(partition);
                counter.count(alertLog);
            } catch (Exception e) {
                LOGGER.error("处理kafka消息异常，消息为：{}", json, e);
            }
        });
    }

    /**
     * 解析日志数据，并返回AlertLog
     * message示例：2021-09-23 09:56:48.286 [rule-service-6d8cb4776f-5gphg] [qtp247574430-1492] WARN [com.brcloud.rule.service.CachedRuleEngineServiceImpl] -ALERT_LEVEL:WARN ERROR_CODE:BR-LOG-10001 MESSAGE:responseTime:200 ,swift_number:1178
     * @param json
     * @return alertLog
     */
    public AlertLog parse(String json) {
        JSONObject jsonObject = JSONObject.parseObject(json);
        //解析hostname
        String hostname = jsonObject.containsKey(FIELD_HOST) && jsonObject.getJSONObject(FIELD_HOST).containsKey(FIELD_HOST_NAME) ?
                jsonObject.getJSONObject(FIELD_HOST).getString(FIELD_HOST_NAME) : VALUE_NONE_HOSTNAME;

        String message = jsonObject.getString(FIELD_MESSAGE);
        Matcher messageMatcher = matchMessage(message);
        String timestamp = Objects.nonNull(messageMatcher) ? messageMatcher.group(1) : VALUE_NONE_TIMESTAMP;
        String msg = Objects.nonNull(messageMatcher) ? messageMatcher.group(6) : VALUE_NONE_MSG;

        Message msgObj = getRealMessage(message, msg);

        String source = jsonObject.getString(FIELD_SOURCE);
        //提取pod
        int lastSplit = source.lastIndexOf(SPITE);
        Assert.state(lastSplit != -1, "日志中的source字段路径不符合规范，解析失败！");

        int podFirst = source.lastIndexOf(SPITE, lastSplit - 1);
        Assert.state(podFirst != -1, "日志中的source字段路径不符合规范，解析失败！");

        String pod = source.substring(podFirst + 1, lastSplit);

        //提取service
        int serviceFirst = source.lastIndexOf(SPITE, podFirst - 1);
        Assert.state(serviceFirst != -1, "日志中的source字段路径不符合规范，解析失败！");

        String service = source.substring(serviceFirst + 1, podFirst);

        //提取子环境

        JSONObject fields = jsonObject.getJSONObject(FIELD_FIELDS);
        String subEnv = Objects.nonNull(fields) ? fields.getString(FIELD_SUB_ENV) : null;

        return AlertLog.builder()
                .logTime(DateUtils.parse("yyyy-MM-dd HH:mm:ss.SSS", timestamp))
                .alertName(msgObj.getMsgContent())
                .level(msgObj.getLevel())
                .subject(msgObj.getSubject())
                .receiver(msgObj.getReceiver())
                .message(msgObj.getOriginalMsg())
                .errorCode(msgObj.getErrorCode())
                .service(service)
                .podName(pod)
                .hostname(hostname)
                .subEnv(subEnv)
                .build();
    }

    private Message getRealMessage(String message, String msg) {
        Matcher msgContentMatcher = matchMessageContent(msg);
        //如果是原来的日志格式，按照原来方式解析，否则使用新正则解析
        if (Objects.nonNull(msgContentMatcher)) {
            return getMessage(msgContentMatcher, message);
        }

        //匹配subject
        msgContentMatcher = matchMessageSubjectContent(msg);
        if (Objects.nonNull(msgContentMatcher)) {
            return getSubjectMessage(msgContentMatcher, message);
        }

        //匹配接收人
        msgContentMatcher = matchMessageReceiverContent(msg);
        if (Objects.nonNull(msgContentMatcher)) {
            return getReceiverMessage(msgContentMatcher, message);
        }

        //匹配subject和接收人
        msgContentMatcher = matchMessageSubjectReceiverContent(msg);
        if (Objects.nonNull(msgContentMatcher)) {
            return getSubjectReceiverMessage(msgContentMatcher, message);
        }

        LOGGER.error("该日志内容解析错误，内容为：{}。", msg);
        throw new IllegalArgumentException("解析日志内容错误，内容为：" + msg);
    }

    private Message getMessage(Matcher msgContentMatcher, String message) {
        Message result = Message.builder()
                .level(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(1) : VALUE_NONE_LEVEL)
                .errorCode(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(2) : VALUE_NONE_ERROR_CODE)
                .msgContent(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(3) : VALUE_NONE_MSG_CONTENT)
                .build();
        //替换message中的打标签部分
        Matcher matcher = messageContentPattern.matcher(message);
        //$符合特殊处理
        result.setOriginalMsg(encode(matcher.replaceFirst(decode(result.getMsgContent()))));
        return result;
    }

    private Message getSubjectMessage(Matcher msgContentMatcher, String message) {
        Message result = Message.builder()
                .level(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(1) : VALUE_NONE_LEVEL)
                .errorCode(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(2) : VALUE_NONE_ERROR_CODE)
                .subject(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(3) : VALUE_NONE_SUBJECT_CONTENT)
                .msgContent(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(4) : VALUE_NONE_MSG_CONTENT)
                .build();
        //替换message中的打标签部分
        Matcher matcher = messageContentContainsSubjectPattern.matcher(message);
        //$符合特殊处理
        result.setOriginalMsg(encode(matcher.replaceFirst(decode(result.getMsgContent()))));
        return result;
    }

    private Message getReceiverMessage(Matcher msgContentMatcher, String message) {
        Message result = Message.builder()
                .level(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(1) : VALUE_NONE_LEVEL)
                .errorCode(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(2) : VALUE_NONE_ERROR_CODE)
                .receiver(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(3) : VALUE_NONE_RECEIVER_CONTENT)
                .msgContent(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(4) : VALUE_NONE_MSG_CONTENT)
                .build();
        //替换message中的打标签部分
        Matcher matcher = messageContentContainsReceiverPattern.matcher(message);
        //$符合特殊处理
        result.setOriginalMsg(encode(matcher.replaceFirst(decode(result.getMsgContent()))));
        return result;
    }

    private Message getSubjectReceiverMessage(Matcher msgContentMatcher, String message) {
        Message result = Message.builder()
                .level(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(1) : VALUE_NONE_LEVEL)
                .errorCode(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(2) : VALUE_NONE_ERROR_CODE)
                .subject(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(3) : VALUE_NONE_SUBJECT_CONTENT)
                .receiver(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(4) : VALUE_NONE_RECEIVER_CONTENT)
                .msgContent(Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(5) : VALUE_NONE_MSG_CONTENT)
                .build();
        //替换message中的打标签部分
        Matcher matcher = messageContentContainsSubjectReceiverPattern.matcher(message);
        //$符合特殊处理
        result.setOriginalMsg(encode(matcher.replaceFirst(decode(result.getMsgContent()))));
        return result;
    }
    private String decode(String source) {
        return source.replaceAll("\\$", RDS_CHAR_DOLLAR);
    }

    private String encode(String source) {
        return source.replaceAll(RDS_CHAR_DOLLAR,"\\$");
    }

    private final Pattern messagePattern = Pattern.compile("(.[0-9]+-[0-9]+-[0-9]+\\s[0-9]+:[0-9]+:[0-9]+.[0-9]+) \\[(.[\\w.-]+)\\] \\[(.[\\S]+)\\] (.[\\S]+)\\s+\\[(.[\\S]+)\\] -(.*)");
    private final Pattern messageContentPattern = Pattern.compile("ALERT_LEVEL:(.[a-z0-9A-Z-]+) ERROR_CODE:(.[a-z0-9A-Z-]+) MESSAGE:(.*)");
    private final Pattern messageContentContainsSubjectPattern = Pattern.compile("ALERT_LEVEL:(.[a-z0-9A-Z-]+) ERROR_CODE:(.[a-z0-9A-Z-]+) SUBJECT:(.\\S+) MESSAGE:(.*)");

    //包括接收人
    private final Pattern messageContentContainsReceiverPattern = Pattern.compile("ALERT_LEVEL:(.[a-z0-9A-Z-]+) ERROR_CODE:(.[a-z0-9A-Z-]+) RECEIVER:(.\\S+) MESSAGE:(.*)");
    private final Pattern messageContentContainsSubjectReceiverPattern = Pattern.compile("ALERT_LEVEL:(.[a-z0-9A-Z-]+) ERROR_CODE:(.[a-z0-9A-Z-]+) SUBJECT:(.\\S+) RECEIVER:(.\\S+) MESSAGE:(.*)");

    private Matcher matchMessageContent(String msg) {
        Matcher matcher = messageContentPattern.matcher(msg);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessageSubjectContent(String msg) {
        Matcher matcher = messageContentContainsSubjectPattern.matcher(msg);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessageReceiverContent(String msg) {
        Matcher matcher = messageContentContainsReceiverPattern.matcher(msg);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessageSubjectReceiverContent(String msg) {
        Matcher matcher = messageContentContainsSubjectReceiverPattern.matcher(msg);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessage(String message) {
        Matcher matcher = messagePattern.matcher(message);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    @Data
    @Builder
    private static class Message {
        private String level;
        private String errorCode;
        private String subject;
        private String receiver;
        private String msgContent;
        //即原始日志去掉标签后的消息
        private String originalMsg;
    }
}
